#include "zmqSocket.h"
#include "reaC++.h"
#include <QJsonDocument>

using namespace std::chrono_literals;
/*static bool inited_rep = false;
static bool inited_req = false;
static rea::regPip<QJsonObject> init_rep([](rea::stream<QJsonObject>* aInput) {
    static zmq::context_t context{1};
    static zmq::socket_t socket{context, zmq::socket_type::rep};
    if (!inited_rep){
        socket.bind("tcp://*:5555");
        inited_rep = true;
        while(1){
            zmq::message_t request;
            socket.recv(request, zmq::recv_flags::none);
            std::cout << "Rep Received " << request.to_string() << std::endl;

            const std::string data{"World"};
            std::cout << "Rep Send " << data << std::endl;
            socket.send(zmq::buffer(data), zmq::send_flags::none);

            std::this_thread::sleep_for(1s);
        }
    }


}, rea::Json("name", "testZMQRep", "thread", 100));

static rea::regPip<QJsonObject> init_req([](rea::stream<QJsonObject>* aInput) {
    static zmq::context_t context{1};
    static zmq::socket_t socket{context, zmq::socket_type::req};
    if (!inited_req){
        socket.connect("tcp://localhost:5555");
        inited_req = true;
    }

    const std::string data{"Hello"};
    std::cout << "Req Send " << data << std::endl;
    socket.send(zmq::buffer(data), zmq::send_flags::none);

    zmq::message_t reply{};
    socket.recv(reply, zmq::recv_flags::none);
    std::cout << "Req Received " << reply.to_string() << std::endl;
}, rea::Json("name", "testZMQReq", "thread", 101));

static rea::regPip<QJsonObject> init_pub([](rea::stream<QJsonObject>* aInput) {
    zmq::context_t context{1};
    zmq::socket_t socket{context, zmq::socket_type::pub};
    socket.bind("tcp://*:5556");
    const std::string data{"Hello"};
    std::this_thread::sleep_for(1s);
    std::cout << "Pub Send " << data << std::endl;
    socket.send(zmq::str_buffer("PubSub"), zmq::send_flags::sndmore);
    socket.send(zmq::buffer(data), zmq::send_flags::none);
}, rea::Json("name", "testZMQPub", "thread", 102));

static rea::regPip<QJsonObject> init_sub([](rea::stream<QJsonObject>* aInput) {
    zmq::context_t context{1};
    zmq::socket_t socket{context, zmq::socket_type::sub};
    socket.connect("tcp://localhost:5556");
    socket.set(zmq::sockopt::subscribe, "PubSub");
    std::vector<zmq::message_t> recv_msgs;
    zmq::recv_result_t result = zmq::recv_multipart(socket, std::back_inserter(recv_msgs));
    std::cout << "Sub Received ";
    for (auto i = recv_msgs.begin(); i != recv_msgs.end(); ++i)
        std::cout << "\"" << i->to_string() << "\"";
    std::cout << std::endl;
}, rea::Json("name", "testZMQSub", "thread", 103));

static rea::regPip<QJsonObject> init_routerdealer([](rea::stream<QJsonObject>* aInput) {
    zmq::context_t context{1};
    zmq::socket_t socket{context, zmq::socket_type::dealer};
    socket.connect("tcp://127.0.0.1:5557");
    const std::string data{"Hello"};
    std::cout << "Dealer Send " << data << std::endl;
    socket.send(zmq::buffer(data), zmq::send_flags::none);

    std::vector<zmq::message_t> recv_msgs;
    zmq::recv_result_t result = zmq::recv_multipart(socket, std::back_inserter(recv_msgs));
    std::cout << "Dealer Received ";
    for (auto i = recv_msgs.begin(); i != recv_msgs.end(); ++i)
        std::cout << "\"" << i->to_string() << "\"";
    std::cout << std::endl;
}, rea::Json("name", "testZMQRouterDealer", "thread", 104));

static rea::regPip<QJsonObject> init_routerdealer3([](rea::stream<QJsonObject>* aInput) {
    zmq::context_t context{1};
    zmq::socket_t socket{context, zmq::socket_type::dealer};
    socket.connect("tcp://127.0.0.1:5557");
    const std::string data{"Hello"};
    std::cout << "Dealer2 Send " << data << std::endl;
    socket.send(zmq::buffer(data), zmq::send_flags::none);

    std::vector<zmq::message_t> recv_msgs;
    zmq::recv_result_t result = zmq::recv_multipart(socket, std::back_inserter(recv_msgs));
    std::cout << "Dealer2 Received ";
    for (auto i = recv_msgs.begin(); i != recv_msgs.end(); ++i)
        std::cout << "\"" << i->to_string() << "\"";
    std::cout << std::endl;
}, rea::Json("name", "testZMQRouterDealer3", "thread", 106));

static rea::regPip<QJsonObject> init_routerdealer2([](rea::stream<QJsonObject>* aInput) {
    zmq::context_t context{1};
    zmq::socket_t socket{context, zmq::socket_type::router};
    socket.bind("tcp://127.0.0.1:5557");

    while(1){

        std::vector<zmq::message_t> recv_msgs;
        zmq::recv_result_t result = zmq::recv_multipart(socket, std::back_inserter(recv_msgs));
        std::cout << "Router Received ";
        for (auto i = recv_msgs.begin(); i != recv_msgs.end(); ++i)
            std::cout << "\"" << i->to_string() << "\"";
        std::cout << std::endl;

        const std::string data{"World"};
        std::cout << "Router Send " << data << std::endl;
        std::cout << recv_msgs[0].to_string() << std::endl;
        socket.send(recv_msgs[0], zmq::send_flags::sndmore);
        socket.send(zmq::buffer(data), zmq::send_flags::none);
        std::this_thread::sleep_for(1s);
    }

}, rea::Json("name", "testZMQRouterDealer2", "thread", 105));

static rea::regPip<QJsonObject> init_push([](rea::stream<QJsonObject>* aInput) {
    zmq::context_t context{1};
    zmq::socket_t socket{context, zmq::socket_type::push};
    socket.bind("tcp://127.0.0.1:5558");
    std::this_thread::sleep_for(1s);
    const std::string data{"Hello"};
    std::cout << "Push Send" << data << std::endl;
    socket.send(zmq::buffer(data), zmq::send_flags::none);
}, rea::Json("name", "testZMQPush", "thread", 106));

static rea::regPip<QJsonObject> init_pull([](rea::stream<QJsonObject>* aInput) {
    zmq::context_t context{1};
    zmq::socket_t socket{context, zmq::socket_type::pull};
    socket.connect("tcp://127.0.0.1:5558");

    zmq::message_t request;
    socket.recv(request, zmq::recv_flags::none);
    std::cout << "Pull Received " << request.to_string() << std::endl;
}, rea::Json("name", "testZMQPull", "thread", 107));
*/
zmqSocket::zmqSocket(const QJsonObject& aConfig, zmq::context_t& aContext){
    address = aConfig.value("ip").toString();
    //qDebug() << aConfig;
    auto ip = address.toStdString();
    auto tp = aConfig.value("type").toString();
    if (tp == "rep"){
        sock = std::make_unique<zmq::socket_t>(aContext, zmq::socket_type::rep);
        sock->bind(ip);
    }else if (tp == "req"){
        sock = std::make_unique<zmq::socket_t>(aContext, zmq::socket_type::req);
        sock->connect(ip);
        conn = true;
    }else if (tp == "pub"){
        sock = std::make_unique<zmq::socket_t>(aContext, zmq::socket_type::pub);
        sock->bind(ip);
        recv = false;
    }else if (tp == "sub"){
        sock = std::make_unique<zmq::socket_t>(aContext, zmq::socket_type::sub);
        sock->connect(ip);
        conn = true;
        sock->set(zmq::sockopt::subscribe, aConfig.value("filter").toString().toStdString());
    }else if (tp == "push"){
        sock = std::make_unique<zmq::socket_t>(aContext, zmq::socket_type::push);
        sock->bind(ip);
        recv = false;
    }else if (tp == "pull"){
        sock = std::make_unique<zmq::socket_t>(aContext, zmq::socket_type::pull);
        sock->connect(ip);
        conn = true;
    }else if (tp == "router"){
        sock  = std::make_unique<zmq::socket_t>(aContext, zmq::socket_type::router);
        sock->bind(ip);
    }else if (tp == "dealer"){
        sock = std::make_unique<zmq::socket_t>(aContext, zmq::socket_type::dealer);
        sock->connect(ip);
        conn = true;
    }
}

zmqSockets::~zmqSockets(){
    //zsys_shutdown();
  /*  if (!m_sockets_lock.try_lock())
        std::this_thread::sleep_for(std::chrono::microseconds(5));
    for (auto i : m_sockets)
        i->sock->close();
    m_sockets_lock.unlock();

    if (!m_recv_sockets_lock.try_lock())
        std::this_thread::sleep_for(std::chrono::microseconds(5));
    m_recv_sockets.clear();
    m_recv_sockets_lock.unlock();*/
}

zmqSockets::zmqSockets(const QString& aName){
    rea::pipeline::add<int, rea::pipeAsync>([this, aName](rea::stream<int>* aInput){
        if (!m_sockets_lock.try_lock())
            std::this_thread::sleep_for(std::chrono::microseconds(5));
        for (auto i : m_sockets){
            if (i->conn && ((i->sock->handle() != nullptr) != i->isconnected)){
                i->isconnected = i->sock->handle() != nullptr;
                aInput->outs(i->isconnected, aName + "statusZMQ_" + m_sockets.key(i));
            }
        }
        m_sockets_lock.unlock();
        std::this_thread::sleep_for(1s);
        aInput->outs(aInput->data(), "");
    }, rea::Json("name", aName + "statusZMQ", "thread", 102))
            ->next(aName + "statusZMQ")
            ->execute(std::make_unique<rea::stream<int>>());

    rea::pipeline::add<int, rea::pipeAsync>([this, aName](rea::stream<int>* aInput){
        if (!m_recv_sockets_lock.try_lock())
            std::this_thread::sleep_for(std::chrono::microseconds(5));
        if (m_recv_sockets.size()){
            try{
                zmq::poll(&m_recv_sockets[0], m_recv_sockets.size(), 5000);
            }catch(...){
                return;
            }
            if (!m_sockets_lock.try_lock() || !m_sockets_ids_lock.try_lock())
                std::this_thread::sleep_for(std::chrono::microseconds(5));
            for (auto j : m_recv_sockets){
              if (j.revents & ZMQ_POLLIN) {
                std::vector<zmq::message_t> recv_msgs;
                auto id = m_sockets_ids.value(j.socket);
                auto soc = m_sockets.value(id);
                zmq::recv_multipart(*soc->sock.get(), std::back_inserter(recv_msgs));
                if (recv_msgs.size()){
                    auto dt = QJsonDocument::fromJson(QByteArray::fromStdString(recv_msgs.back().to_string())).object();
                    //qDebug() << dt;
                        //std::cout << id.toStdString() << std::endl;
                    auto ot = aInput->outs(dt, aName + "receiveZMQ_" + id);
                    if (recv_msgs.size() > 1)
                        ot->var("connect", recv_msgs.front().to_string());
                }
              }
            }
            m_sockets_lock.unlock();
            m_sockets_ids_lock.unlock();
        }else
            std::this_thread::sleep_for(1s);
        m_recv_sockets_lock.unlock();
        aInput->outs(aInput->data(), "");
    }, rea::Json("name", aName + "receiveZMQ", "thread", 100))
            ->next(aName + "receiveZMQ")
            ->execute(std::make_unique<rea::stream<int>>());

    rea::pipeline::add<QJsonObject>([this](rea::stream<QJsonObject>* aInput){
        if (!m_sockets_lock.try_lock())
            std::this_thread::sleep_for(std::chrono::microseconds(5));
        auto dt = aInput->data();
        auto soc = m_sockets.value(dt.value("id").toString());
        if (soc){
            auto cnts = dt.value("data").toArray();
            //if (dt.value("id") == "pub")
            //    qDebug() << cnts;
            auto cnt = aInput->varData<std::string>("connect");
            if (cnt != "")
                soc->sock->send(zmq::buffer(cnt), zmq::send_flags::sndmore);
            for (auto i = 0; i < cnts.size(); ++i){
                soc->sock->send(zmq::buffer(cnts[i].toString().toStdString()), i == cnts.size() - 1 ? zmq::send_flags::none : zmq::send_flags::sndmore);
            }
        }
        m_sockets_lock.unlock();
    }, rea::Json("name", aName + "postZMQ", "thread", 101));
}

zmqSocket::~zmqSocket(){
    sock->close();
}

void zmqSockets::add(const QJsonObject& aConfig){
    //qDebug() << "add socs";
    if (!m_sockets_lock.try_lock() || !m_recv_sockets_lock.try_lock() || !m_sockets_ids_lock.try_lock())
        std::this_thread::sleep_for(std::chrono::microseconds(5));
    for (auto i : aConfig.keys()){
        auto cfg = aConfig.value(i).toObject();
        auto soc = std::make_shared<zmqSocket>(cfg, m_context);
        //qDebug() << i;
        m_sockets.insert(i, soc);
        auto poll_soc = static_cast<void *>(*soc->sock.get());
        if (soc->recv){
            zmq::pollitem_t tmpItem;
            tmpItem.socket = poll_soc;
            tmpItem.fd = 0;
            tmpItem.events = ZMQ_POLLIN;
            tmpItem.revents = 0;
            m_recv_sockets.push_back(tmpItem);
        }
       // std::cout << poll_soc << std::endl;
        m_sockets_ids.insert(poll_soc, i);
    }
    m_sockets_lock.unlock();
    m_recv_sockets_lock.unlock();
    m_sockets_ids_lock.unlock();
}

static rea::regPip<QJsonObject> init_main([](rea::stream<QJsonObject>* aInput) {
    /*rea::pipeline::run("testZMQRep", QJsonObject());
    rea::pipeline::run("testZMQReq", QJsonObject());
    rea::pipeline::run("testZMQPub", QJsonObject());
    rea::pipeline::run("testZMQSub", QJsonObject());
    rea::pipeline::run("testZMQRouterDealer", QJsonObject());
    rea::pipeline::run("testZMQRouterDealer2", QJsonObject());
    rea::pipeline::run("testZMQRouterDealer3", QJsonObject());
    rea::pipeline::run("testZMQPush", QJsonObject());
    rea::pipeline::run("testZMQPull", QJsonObject());*/
    static bool inited = false;
    if (!inited){
        static zmqSockets socs("test");
        socs.add(rea::Json(
                            "1", rea::Json("type", "req",
                                           "ip", "tcp://localhost:5555"),
                            "2", rea::Json("type", "rep",
                                           "ip", "tcp://*:5555")
                            ));
        rea::pipeline::add<bool>([](rea::stream<bool>* aInput){
            qDebug() << "1 connected: " << aInput->data();
        }, rea::Json("name", "teststatusZMQ_1"));
        rea::pipeline::add<QJsonObject>([](rea::stream<QJsonObject>* aInput){
            qDebug() << "1 received: " << aInput->data();
        }, rea::Json("name", "testreceiveZMQ_1"));
        rea::pipeline::add<QJsonObject>([](rea::stream<QJsonObject>* aInput){
            qDebug() << "2 received: " << aInput->data();
            QString dt = QJsonDocument(rea::Json("world", "hello")).toJson();
            aInput->outs(rea::Json("id", "2", "data", rea::JArray(dt)), "postZMQ");
        }, rea::Json("name", "testreceiveZMQ_2"));

        socs.add(rea::Json("3", rea::Json("type", "pub",
                                          "ip", "tcp://*:5556"),
                           "4", rea::Json("type", "sub",
                                          "ip", "tcp://127.0.0.1:5556",
                                          "filter", "dpst")));
        rea::pipeline::add<QJsonObject>([](rea::stream<QJsonObject>* aInput){
            qDebug() << "4 received: " << aInput->data();
        }, rea::Json("name", "testreceiveZMQ_4"));

        socs.add(rea::Json("5", rea::Json("type", "dealer",
                                          "ip", "tcp://127.0.0.1:5557"),
                           "6", rea::Json("type", "router",
                                          "ip", "tcp://*:5557")));
        rea::pipeline::add<QJsonObject>([](rea::stream<QJsonObject>* aInput){
            qDebug() << "5 received: " << aInput->data();
        }, rea::Json("name", "testreceiveZMQ_5"));
        rea::pipeline::add<QJsonObject>([](rea::stream<QJsonObject>* aInput){
            qDebug() << "6 received: " << aInput->data();
            QString dt = QJsonDocument(rea::Json("world", "hello")).toJson();
            aInput->outs(rea::Json("id", "6", "data", rea::JArray(dt)), "postZMQ");
        }, rea::Json("name", "testreceiveZMQ_6"));

        socs.add(rea::Json("7", rea::Json("type", "push",
                                          "ip", "tcp://*:5558"),
                           "8", rea::Json("type", "pull",
                                          "ip", "tcp://127.0.0.1:5558")));
        rea::pipeline::add<QJsonObject>([](rea::stream<QJsonObject>* aInput){
            qDebug() << "8 received: " << aInput->data();
        }, rea::Json("name", "testreceiveZMQ_8"));
        inited = true;
    }
    QString dt = QJsonDocument(rea::Json("hello", "world")).toJson();
    //rea::pipeline::run<QJsonObject>("testpostZMQ", rea::Json("id", "1", "data", rea::JArray(dt)));
    rea::pipeline::run<QJsonObject>("testpostZMQ", rea::Json("id", "3", "data", rea::JArray("dpst", dt)));
    //rea::pipeline::run<QJsonObject>("testpostZMQ", rea::Json("id", "5", "data", rea::JArray(dt)));
    //rea::pipeline::run<QJsonObject>("testpostZMQ", rea::Json("id", "7", "data", rea::JArray(dt)));

}, rea::Json("name", "testZMQ"));
